package gbench.appdemo.priceline.app;

import org.junit.jupiter.api.Test;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import gbench.appdemo.priceline.indicator.OHLC;
import gbench.appdemo.priceline.model.PeriodPriceLineModel;
import gbench.appdemo.priceline.model.PriceLineDatabase;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.Jdbc;
import gbench.commonApp.data.DataApp;
import static gbench.common.tree.LittleTree.*;
import static gbench.common.tree.LittleTree.IRecord.*;

/**
 * 
 * @author gbench
 *
 */
public class JunitKLineModel3 {
    
    public class MyData extends DataApp{
        public List<IRecord> sqlquery(final String sql) {
            cfg.set("url", "jdbc:mysql://localhost:3306/futures?serverTimezone=GMT%2B8"); reload();
            return this.jdbc.sql2records(sql);
        }
        
        public IRecord config() {
            cfg.set("url", "jdbc:mysql://localhost:3306/futures?serverTimezone=GMT%2B8"); reload();
            return this.cfg;
        }
        
        public IRecord config(String dbname) {
            return cfg.derive("url", MFT("jdbc:mysql://localhost:3306/{0}?serverTimezone=GMT%2B8",dbname));
        }
    }
    
    /**
     * 转换成 列表元素
     * @param <T>
     * @param <U>
     * @param mm
     * @param f
     * @return
     */
    public static <T,U> List<U> m2l(final Map<?,T>mm, final Function<T,U> mapper){
        return mm.values().stream().map(mapper).collect(Collectors.toList());
    }
    
    /**
     * 
     * @author gbench
     *
     * @param <T>
     * @param <U>
     */
    public static  class CHMap<K,V> extends ConcurrentHashMap<K,V>{
        private static final long serialVersionUID = 1L;
        
        /**
         * 
         * @param <U>
         * @param k
         * @param mapper
         * @return
         */
        public <U> U get(K k,Function<V,U>mapper) {
            final var v = this.get(k);
            return v==null?null:mapper.apply(v);
        }
        
        /**
         * 
         * @param <U>
         * @param cs
         * @param mapper
         */
        public <U> void forEach(final BiConsumer<K,U> cs,final Function<V,U>mapper) {
            this.forEach((k,v)->{
                cs.accept(k,mapper.apply(v));
            });
        }
    }
    
    /**
     * 键值
     * @param tkr ticker
     * @param period period
     * @return Key
     */
    final String K(final String tkr,final Number period) {
        return MFT("{0}/{1,number,#}",tkr,period);
    }
    
    @Test
    public void foo() {
        final var dataApp = new MyData();
        final var redisData = new CHMap<String,List<IRecord>>();// 数据缓存,模拟redis数据库
        final var klineDatabase = new CHMap<String,Map<Long,OHLC>>();// K线数据
        final var latch = new CountDownLatch(1);
        final var pricelineDatabase = Jdbc.newInstance(PriceLineDatabase.class,dataApp.config("db1"));
        final var pool = Executors.newFixedThreadPool(10);
        final var disruptor = new Disruptor<IRecord>(()->REC(),1024*1024, Executors.defaultThreadFactory(),ProducerType.MULTI,new YieldingWaitStrategy());
        disruptor.handleEventsWith(new EventHandler<IRecord>() {
            final List<IRecord> localCache = new LinkedList<>();
            @Override
            public void onEvent(IRecord event, long sequence, boolean endOfBatch) throws Exception {
                if(event.size()<1) {latch.countDown(); return;}
                localCache.add(event);
                if(endOfBatch) {
                    localCache.stream().collect(Collectors.groupingBy(e->e.str("ticker"))).forEach((ticker,rr)->{
                        System.out.println(MFT("\nid:{0},size:{1}",Thread.currentThread().getId(),localCache.size()));
                        final var mm = new HashMap<Number,IRecord>();
                        rr.forEach(r-> mm.compute(r.i4("index"), (k,v)->v==null ? r: v.i4("time")>r.i4("time") ? v:r ));
                        mm.forEach((k,m)->{
                            System.out.println(MFT("{0}",m));
                            final var tableName = MFT("t_priceline_{0}",m.get("ticker"));
                            //if(!pricelineDatabase.exists(tableName))pricelineDatabase.createTable(tableName);
                            pricelineDatabase.persist(tableName, m.str("ticker"), m.lng("period"), 
                                m.lng("index"), m.date("time"), m.dbl("open"), m.dbl("high"), 
                                m.dbl("low"), m.dbl("close"), m.lng("volume"), m.lng("volume0"));
                        });
                    });
                    
                    localCache.clear();
                }
            }
        });
        disruptor.start();
        
        Stream.of("mpg001","mpg002","mpg003","mpg004") // 生成合约代码
        .parallel()
        .map(ticker->KLineModel.newInstance(REC(
            "ticker",ticker,
            "auto_create",true,
            "periods",A(60000,300000,600000,3000000),
            "redis_data",redisData,
            "jdbc",Jdbc.M("url", "jdbc:mysql://localhost:3306/futures?serverTimezone=GMT%2B8", 
                "driver", "com.mysql.cj.jdbc.Driver","user", "root","password", "123456"), // jdbc 数据库配置
            "inner_callback",(BiConsumer<PeriodPriceLineModel<OHLC>.PeriodPriceLineProcessor,OHLC>) (p,ohlc)->{
                pool.execute(()->{// 转移到数据库
                    disruptor.getRingBuffer().publishEvent((rec,seq)->
                        rec.add(IRecord.OBJ2REC(ohlc)).add(REC("ticker",p.getTicker(),"period",p.getPeriod()))
                    );// 加入到ringbuffer
                });
                klineDatabase.compute(K(p.getTicker(),p.getPeriod()), (k, k2ohlcs) -> { // K线数据库的读写 (tiker
                    if(k2ohlcs==null) k2ohlcs = new ConcurrentHashMap<Long,OHLC>(); // 寻找ticker map 数据
                    k2ohlcs.compute(ohlc.getIndex(), (idx,_ohlc) -> _ohlc == null 
                        ? ohlc: _ohlc.getTime().isBefore(ohlc.getTime()) ? ohlc : ohlc);// index
                    return k2ohlcs;
                });// ticker,period
                //System.out.println(ohlc);
            }// inner_callback
        ))).forEach( klm -> {
            dataApp.sqlquery("SELECT time, price, volume FROM ohlc1.t_deriv_eg2001 limit 10000").forEach(klm::handle);
            klm.flush();// 刷新数据到数据库，持久保存
        });
        
        disruptor.getRingBuffer().publishEvent((rec,seq)->rec.add(REC()));// 结束事件
        
        // 倾倒redis缓存数据
        System.out.println("\n----------------------处理日志----------------------------");
        redisData.forEach((k,recs)->System.out.println(MFT("\n{0}:\n{1}",k,IRecord.FMT(recs.subList(0, Math.min(recs.size(),1000))))));
        System.out.println("\n----------------------日志大小----------------------------");
        redisData.forEach((k,recs)->System.out.println(k+":"+recs.size()));
        System.out.println("\n----------------------K线数据----------------------------");
        klineDatabase.forEach((k,priceline)->{
            System.out.println(IRecord.FMT(priceline));
        },v->m2l(v,IRecord::OBJ2REC));
        
        try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}
        
    }
    

}
